package cn.jolyoulu.mqtt.config.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Import;

import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;


@Slf4j
@Import(MqttProperties.class)
public class MqttClientPool {

    /**
     * jwt超时时间
     */
    public static final Integer jwtTimeOutDay = 365;

    /**
     * jwt密钥
     */
    public static final String jwtSecret = "test";

    /**
     * 连接池
     */
    private static BlockingQueue<MqttClient> pool;

    /**
     * 锁
     */
    private final static Object lock = new Object();


    public MqttClientPool(MqttProperties properties) {
        pool = new LinkedBlockingQueue<>(properties.getPoolSize() * 2); //预留一倍空间放那些掉线了的客户端，掉线了的客户端只要有回收时才会被移除
        //初始化连接池
        for (int i = 0; i < properties.getPoolSize(); i++) {
            String clientId = properties.getNode() + "-MqttClientPool-" + (i + 1);
            OnMessageCallback callback = new OnMessageCallback(clientId, properties);
            try {
                // 初始化一个客户端
                MqttClient client = createClient(clientId, properties.getBroker(), jwtSecret, callback);
                pool.offer(client);
            } catch (MqttException e) {
                log.error("MQTT客户端[" + clientId + "]创建失败", e);
            }
        }
    }

    /**
     * 从连接池中获取一个 MqttClient
     * 如果连接池连接耗尽，等待5后仍无可以连接，则返回null
     * @return MqttClient
     */
    public static MqttClient getClient() {
        synchronized (lock){
            try {
                return pool.poll(5, TimeUnit.SECONDS);//阻塞5秒获取客户端
            } catch (InterruptedException e) {
                log.error("MqttClientPool获取客户端失败，连接池资源耗尽", e);
                return null;
            }
        }
    }

    /**
     * 归还一个连接给连接池
     * 如果连接是null或者连接关闭的会直接丢弃
     * @param client
     */
    public static void returnClient(MqttClient client) {
        if (client != null && client.isConnected()) {
            pool.offer(client);
        }
    }

    /**
     * 删除连接池中的一个连接
     * @param clientId
     */
    public static void removeClient(String clientId){
        synchronized (lock){
            try {
                //移除
                Iterator<MqttClient> iterator = pool.iterator();
                while (iterator.hasNext()) {
                    MqttClient next = iterator.next();
                    if (next.getClientId().equals(clientId)) {
                        next.close(true);
                        iterator.remove(); //删除元素
                        break;
                    }
                }
            } catch (MqttException e) {
                log.error("MqttClientPool移除客户端失败", e);
            }
        }
    }

    /**
     * 创建一个客户端
     * 自动重连是关闭的，需要手动处理
     * @param clientId  客户端id
     * @param broker    连接地址
     * @param jwtSecret token配置
     * @param callback  设置回调
     * @return MqttClient
     * @throws MqttException 异常
     */
    public static MqttClient createClient(String clientId, String broker,
                                          String jwtSecret, MqttCallback callback) throws MqttException {
        // 初始化一个客户端
        MemoryPersistence persistence = new MemoryPersistence();
        MqttClient client = new MqttClient(broker, clientId, persistence);
        // MQTT 连接选项
        MqttConnectOptions connOpts = new MqttConnectOptions();
        // 账号密码
        String adminJwtToken = MqttTokenUtils.createAdminJwtToken(clientId, jwtSecret, jwtTimeOutDay);
        connOpts.setUserName(clientId);
        connOpts.setPassword(adminJwtToken.toCharArray());
        // 关闭清理会话
        connOpts.setCleanSession(true);
        // 开启自动重连，手动处理
        connOpts.setAutomaticReconnect(false);
        // 设置消息回调
        client.setCallback(callback);
        // 建立连接
        log.info("MQTT客户端[{}]正在与服务端建立连接 ==> {}", clientId, broker);
        client.connect(connOpts);
        log.info("MQTT客户端[{}]端连接成功", clientId);
        return client;
    }

    public static class OnMessageCallback implements MqttCallback {

        /**
         * 客户端id
         */
        private final String clientId;

        /**
         * 环境
         */
        private final MqttProperties properties;

        public OnMessageCallback(String clientId, MqttProperties properties) {
            this.clientId = clientId;
            this.properties = properties;
        }

        public void connectionLost(Throwable cause) {
            // 连接丢失后，一般在这里面进行重连
            log.error("MQTT客户端[" + clientId + "] 与MQTT连接失败");
            OnMessageCallback callback = new OnMessageCallback(clientId, properties);
            //无限重试，直到成功
            while (true) {
                try {
                    MqttClient client = createClient(clientId, properties.getBroker(), jwtSecret, callback);
                    MqttClientPool.removeClient(clientId);
                    MqttClientPool.returnClient(client);
                    break;
                } catch (MqttException e) {
                    log.error("MQTT客户端[" + clientId + "]创建失败", e);
                }
            }
        }

        public void messageArrived(String topic, MqttMessage message) {
            // subscribe后得到的消息会执行到这里面
        }

        public void deliveryComplete(IMqttDeliveryToken token) {
            log.info("MQTT客户端[{}]消息发送成功{}", clientId, token.isComplete());
        }
    }
}
